package com.cloud.job.consumer;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.cloud.api.domain.alarm.VehicleAlarm;
import com.cloud.api.domain.job.Alarm;
import com.cloud.api.myenum.AlarmTypeEnum;
import com.cloud.core.exception.MyException;
import com.cloud.job.config.BaiduConfig;
import com.cloud.job.service.IVehicleAlarmService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;

import java.util.ArrayList;
import java.util.List;

/**
 * @author 杨斌
 * 2024-12-2024-12-26
 */
@Slf4j
@Component
public class AlarmConsumer {

    @Resource
    private  BaiduConfig config;

    @Autowired
    private IVehicleAlarmService vehicleAlarmService;



    @KafkaListener(id="${listener.consumer5.id}", topics = "#{'${listener.consumer5.topics}'.split(',')}", groupId = "${listener.consumer5.groupId}", concurrency = "${listener.consumer5.concurrency}")
    public void processAlarmMessage(@Payload List<ConsumerRecord<String, String>> consumerRecords,
                                    Acknowledgment acknowledgment) {
        log.info("Starting to process alarm messages...");

        try {
            long startTime = System.currentTimeMillis(); // 开始计时
            log.info("Polled {} alarm messages ", consumerRecords.size());
            if (consumerRecords.isEmpty()) {
                log.info("No alarm messages to process");
                return; // 返回空列表
            }

            List<VehicleAlarm> vehicleAlarms = new ArrayList<>();
            for (ConsumerRecord<String, String> record : consumerRecords) {
                try {
                    JSONObject jsonObject = JSONUtil.parseObj(record.value());
                    // 验证消息是否包含必要的字段
                    if (!jsonObject.containsKey("vin") || !jsonObject.containsKey("lat") || !jsonObject.containsKey("lng")
                            || !jsonObject.containsKey("level") || !jsonObject.containsKey("alarmType") || !jsonObject.containsKey("localDateTime")) {
                        continue;
                    }
                    // 解析消息
                    Alarm alarm = JSONUtil.toBean(jsonObject, Alarm.class);
                    if (alarm == null) {
                        log.error("Failed to parse alarm message: {}", record.value());
                        continue;
                    }
                    String location = getLocation(alarm.getLat(), alarm.getLng());
                    VehicleAlarm vehicleAlarm = new VehicleAlarm();
                    vehicleAlarm.setAlarmTime(alarm.getLocalDateTime());
                    vehicleAlarm.setVin(alarm.getVin());
                    vehicleAlarm.setAlarmLevel(Integer.valueOf(alarm.getLevel()));
                    vehicleAlarm.setAlarmAddress(location);
                    vehicleAlarm.setAlarmType(AlarmTypeEnum.getByValue(alarm.getAlarmType()));
                    vehicleAlarms.add(vehicleAlarm);
                } catch (Exception e) {
                    log.error("Error parsing alarm messages: {}", record.value(), e);
                }
            }


            if (!vehicleAlarms.isEmpty()) {
                vehicleAlarmService.saveBatch(vehicleAlarms);
                long endTime = System.currentTimeMillis();
                log.info("Inserted {} alarm messages into the database in {} ms", vehicleAlarms.size(), endTime - startTime);
                acknowledgment.acknowledge();
                log.info("Committed offsets for processed alarm messages.");
            } else {
                log.info("No parsed alarm messages to insert into database");
            }

        } catch (Exception e) {
            throw new MyException("Failed to insert alarm data");
        }
    }


    public String getLocation(Long latData,Long lngData){
        String address = "";
        String lat = String.valueOf(latData / 1000000.0);
        String lng = String.valueOf(lngData / 1000000.0);
        String url = UriComponentsBuilder.fromHttpUrl("https://api.map.baidu.com/reverse_geocoding/v3/")
                .queryParam("ak", config.getAk())
                .queryParam("ret_coordtype", config.getRetCoordType())
                .queryParam("coordtype", config.getCoordType())
                .queryParam("extensions_poi", config.getExtensionsPoi())
                .queryParam("extensions_road", config.getExtensionsRoad())
                .queryParam("extensions_town", config.getExtensionsTown())
                .queryParam("output", "json")
                .queryParam("location", lat + "," + lng)
                .toUriString();
        try {
            ResponseEntity<String> response = new RestTemplate().getForEntity(url, String.class);
            JSONObject resultJson = JSONUtil.parseObj(response.getBody());
            if (Integer.parseInt(resultJson.get("status").toString()) != 0) {
                throw new MyException(resultJson.get("message").toString());
            }
            JSONObject result = resultJson.getJSONObject("result");
            if (result.getStr("formatted_address_poi")!= null){
                address = result.getStr("formatted_address_poi");
            }else {
                address = result.getStr("formatted_address");
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        return address;
    }
}
